package com.swak.rabbit;

import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Stream;

import org.springframework.util.Assert;

import com.swak.OS;
import com.swak.rabbit.exception.PublishException;
import com.swak.rabbit.handler.SubscriberBean;
import com.swak.rabbit.message.Message;
import com.swak.rabbit.message.PendingConfirm;
import com.swak.rabbit.retry.RetryPublisher;
import com.swak.utils.Maps;
import com.swak.utils.StringUtils;

import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;

/**
 * 基于 Mq 的事件发布实现
 * 
 * @author lifeng
 */
public class EventBus {

	private static EventBus me = null;
	private volatile boolean inited = false;
	private volatile AtomicBoolean sending = new AtomicBoolean(false);
	private RabbitMQTemplate template;
	private RetryPublisher retryStrategy;
	private Executor executor;
	private Function<RabbitMQTemplate, Boolean> apply;
	private BlockingQueue<QueueMessage> queue = null;

	private EventBus(RabbitMQTemplate template, RetryPublisher strategy, Executor executor,
			Function<RabbitMQTemplate, Boolean> apply) {
		this.template = template;
		this.executor = executor == null ? ForkJoinPool.commonPool() : executor;
		this.retryStrategy = strategy;
		this.queue = new LinkedBlockingDeque<>();
		this.apply = apply;

		if (this.retryStrategy != null) {
			this.template.setConfirmCallback(this.retryStrategy);
			this.template.setReturnCallback(this.retryStrategy);
		}
	}

	/**
	 * 初始化,返回当前是否已初始化
	 */
	public synchronized void init(Consumer<Boolean> register) {
		if (!inited) {
			Optional.of(template).map(t -> this.apply(t)).map(apply).ifPresent(this.delayConsumer(register));
		}
		inited = true;
	}

	/**
	 * 注册成为消费者
	 * 
	 * @param register
	 * @return
	 */
	private Consumer<Boolean> delayConsumer(Consumer<Boolean> register) {
		return (t) -> {
			register.accept(t);
		};
	}

	/**
	 * 默认的队列
	 * 
	 * @param template
	 * @return
	 */
	private RabbitMQTemplate apply(RabbitMQTemplate sender) {

		// 定义服务队列
		sender.exchangeFanoutBindQueue(this.applyModule(Constants.server_channel),
				this.applyModule(this.getServerChannel()), Maps.newHashMap());

		// 定义延迟队列
		Map<String, Object> agruments = Maps.newHashMap();
		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry_channel));
		agruments.put(Constants.x_message_ttl, Constants.dead);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.dead_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry1s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry1s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[0]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry1s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry5s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry5s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[1]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry5s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry10s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry10s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[2]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry10s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry30s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry30s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[3]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry30s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry60s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry60s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[4]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry60s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry120s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry120s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[5]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry120s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry180s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry180s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[6]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry180s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry240s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry240s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[7]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry240s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry300s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry300s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[8]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry300s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry360s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry360s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[9]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry360s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry420s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry420s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[10]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry420s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry480s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry480s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[11]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry480s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry540s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry540s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[12]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry540s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry600s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry600s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[13]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry600s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry1200s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry1200s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[14]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry1200s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry1800s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry1800s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[15]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry1800s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry3600s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry3600s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[16]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry3600s_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry7200s_channel_cus));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry7200s_channel_cus));
		agruments.put(Constants.x_message_ttl, Constants.retrys[17]);
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry7200s_channel), agruments);

		// 定义延迟队列 的消费在队列
		return this.applyCus(sender);
	}

	/**
	 * 默认的队列
	 * 
	 * @param template
	 * @return
	 */
	private RabbitMQTemplate applyCus(RabbitMQTemplate sender) {

		Map<String, Object> agruments = Maps.newHashMap();
		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry1s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry1s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry_channel), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry5s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry5s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry1s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry10s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry10s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry5s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry30s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry30s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry10s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry60s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry60s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry30s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry120s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry120s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry60s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry180s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry180s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry120s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry240s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry240s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry180s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry300s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry300s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry240s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry360s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry360s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry300s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry420s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry420s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry360s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry480s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry480s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry420s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry540s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry540s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry480s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry600s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry600s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry540s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry1200s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry1200s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry600s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry1800s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry1800s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry1200s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry3600s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry3600s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry1800s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.retry7200s_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.retry7200s_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry3600s_channel_cus), agruments);

		agruments.put(Constants.x_dead_letter_exchange, this.applyModule(Constants.fail_channel));
		agruments.put(Constants.x_dead_letter_routing_key, this.applyModule(Constants.fail_channel));
		sender.exchangeDirectBindQueue(this.applyModule(Constants.retry7200s_channel_cus), agruments);

		sender.exchangeDirectBindQueue(this.applyModule(Constants.fail_channel), null);

		return sender;
	}

	private String getServerChannel() {
		return Constants.server_channel + "." + OS.appID();
	}

	private String applyModule(String queue) {
		return this.template.applyModule(queue);
	}

	/**
	 * 注册成为事件消费者
	 * 
	 * @param object
	 */
	public void register(Object object) {
		List<SubscriberBean> subscribers = SubscriberBean.findAllSubscribers(object);
		subscribers.stream().flatMap(subscriber -> {
			return Stream.iterate(0, t -> t + 1).limit(subscriber.getParallel()).map(t -> subscriber);
		}).forEach(subscriber -> {
			template.basicConsume(subscriber.getQueue().equals(Constants.server_channel) ? this.getServerChannel()
					: subscriber.getQueue(), subscriber.getRetry(), subscriber.getPrefetch(), subscriber);
		});
	}

	// 异步发送，提交到任务队列中
	/**
	 * 提交任务
	 * 
	 * @param runnable
	 */
	public void execute(Runnable runnable) {
		this.executor.execute(runnable);
	}

	// 异步发送，消息先提交到队列，通过线程池来小任务的提交到队列
	/**
	 * 队列发送
	 * 
	 * @param exchange
	 * @param routingKey
	 * @param message
	 */
	public void queue(String exchange, String routingKey, Message message) {
		Assert.notNull(message, "Message can not null!");
		this.queue.add(new QueueMessage().setExchange(exchange).setRoutingKey(routingKey).setMessage(message));
		this.prepareTask();
	}

	/**
	 * 添加一次执行任务
	 */
	private void prepareTask() {
		if (sending.compareAndSet(false, true)) {
			this.executor.execute(new QueueRunable());
		}
	}

	// 同步发送消息，保证消息提交到队列

	/**
	 * 发送消息
	 * 
	 * @param exchange
	 * @param routingKey
	 * @param message
	 */
	public void post(String exchange, String routingKey, Object message) {
		Assert.notNull(message, "Message can not null!");
		Message _message = Message.of().object2Payload(message);
		this.post(exchange, routingKey, _message);
	}

	/**
	 * 发送消息 -- String, 获取消息时需要自己转换
	 * 
	 * @param exchange
	 * @param routingKey
	 * @param message
	 */
	public void post(String exchange, String routingKey, String message) {
		Assert.notNull(message, "Message can not null!");
		Message _message = Message.of().setPayload(StringUtils.getBytesUtf8(message));
		this.post(exchange, routingKey, _message);
	}

	/**
	 * 同步的发送消息，将消息提交到任务隊列等待發送
	 * 
	 * @param exchange
	 * @param routingKey
	 * @param message
	 */
	public void post(String exchange, String routingKey, Message message) {
		this.blockSend(exchange, routingKey, message);
	}

	// 将消息提交到队列，需要等待消息队列的返回才能继续执行

	/**
	 * 发送消息
	 * 
	 * @param exchange
	 * @param routingKey
	 * @param message
	 */
	public CompletableFuture<Void> submit(String exchange, String routingKey, Object message) {
		Assert.notNull(message, "Message can not null!");
		Message _message = Message.of().object2Payload(message);
		return this.submit(exchange, routingKey, _message);
	}

	/**
	 * 发送消息 -- String, 获取消息时需要自己转换
	 * 
	 * @param exchange
	 * @param routingKey
	 * @param message
	 */
	public CompletableFuture<Void> submit(String exchange, String routingKey, String message) {
		Assert.notNull(message, "Message can not null!");
		Message _message = Message.of().setPayload(StringUtils.getBytesUtf8(message));
		return this.submit(exchange, routingKey, _message);
	}

	/**
	 * 异步 - 发送消息 - 可以自定义发送重试的次数
	 * 
	 * @param exchange
	 * @param routingKey
	 * @param message
	 */
	public CompletableFuture<Void> submit(String exchange, String routingKey, Message message) {
		return CompletableFuture.runAsync(() -> {
			this.blockSend(exchange, routingKey, message);
		}, executor);
	}

	/**
	 * 阻塞式的消息发送方式（会阻塞调用线程）
	 * 
	 * @param exchange
	 * @param routingKey
	 * @param message
	 * @param retryable
	 */
	@Deprecated
	public void blockSend(String exchange, String routingKey, Message message) {
		message = message.build();
		PendingConfirm pendingConfirm = null;
		boolean retryable = message.getPublishRetrys() > 0;
		Throwable error = null;
		try {
			pendingConfirm = this.template.basicPublish(exchange, routingKey, message, retryable, retryable);
		} catch (Throwable e) {
			error = e;
			pendingConfirm = new PendingConfirm(message.getId());
		}

		// 添加重试
		if (this.retryStrategy != null && pendingConfirm != null && retryable) {
			pendingConfirm.setExchange(exchange);
			pendingConfirm.setRoutingKey(routingKey);
			message.bindPendingConfirm(pendingConfirm);
			this.retryStrategy.add(pendingConfirm);
		}

		// 发送过程中出现错误
		if (error != null) {
			throw new PublishException(error);
		}
	}

	/**
	 * 当前对象
	 * 
	 * @return
	 */
	public static EventBus me() {
		return me;
	}

	/**
	 * 队列消息
	 * 
	 * @author lifeng
	 * @date 2021年3月3日 上午10:28:41
	 */
	@Getter
	@Setter
	@Accessors(chain = true)
	public class QueueMessage {
		private String exchange;
		private String routingKey;
		private Message message;
	}

	/**
	 * 队列发送
	 * 
	 * @author lifeng
	 */
	public class QueueRunable implements Runnable {

		@Override
		public void run() {
			try {
				long now = System.currentTimeMillis();
				while (true) {
					final QueueMessage event = queue.poll();
					if (event == null) {
						break;
					}
					this.doSend(event);
					long dur = System.currentTimeMillis();
					if (dur - now >= 1000) {
						break;
					}
				}
			} finally {
				this.prepareNextTask();
			}
		}

		/**
		 * 特殊情况下使用阻塞式的发送
		 * 
		 * @param event
		 */
		private void doSend(final QueueMessage event) {
			try {
				EventBus.me().blockSend(event.getExchange(), event.getRoutingKey(), event.getMessage());
			} catch (Exception e) {
			}
		}

		/**
		 * 准备下一次的执行
		 */
		private void prepareNextTask() {
			if (sending.compareAndSet(true, false) && queue.peek() != null) {
				prepareTask();
			}
		}
	}

	/**
	 * 构造对象
	 * 
	 * @return
	 */
	public static Builder builder() {
		return new Builder();
	}

	public static class Builder {
		private RabbitMQTemplate template;
		private RetryPublisher strategy;
		private Executor executor;
		private RabbitMqInitialize apply;

		public Builder setTemplate(RabbitMQTemplate template) {
			this.template = template;
			return this;
		}

		public Builder setRetryPublisher(RetryPublisher strategy) {
			this.strategy = strategy;
			return this;
		}

		public Builder setExecutor(Executor executor) {
			this.executor = executor;
			return this;
		}

		public Builder setInitialize(RabbitMqInitialize apply) {
			this.apply = apply;
			return this;
		}

		public EventBus build() {
			EventBus eventBus = new EventBus(template, strategy, executor, apply);
			me = eventBus;
			return eventBus;
		}
	}
}